package com.sh.data.engine.domain.datadev.offline.service.impl;

import cn.hutool.core.convert.Convert;
import com.alibaba.druid.DbType;
import com.alibaba.excel.EasyExcel;
import com.alibaba.excel.ExcelWriter;
import com.alibaba.excel.support.ExcelTypeEnum;
import com.alibaba.excel.write.metadata.WriteSheet;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import com.sh.data.engine.common.constants.Constants;
import com.sh.data.engine.common.enumDefinition.DSType;
import com.sh.data.engine.common.exception.BusinessException;
import com.sh.data.engine.common.util.DateUtil;
import com.sh.data.engine.common.util.DateUtil.DATE_PATTERN;
import com.sh.data.engine.common.util.ScheduleCycleUtil;
import com.sh.data.engine.domain.base.model.PageResult;
import com.sh.data.engine.domain.base.model.enums.Database;
import com.sh.data.engine.domain.base.service.LogService;
import com.sh.data.engine.domain.common.service.DistributedService;
import com.sh.data.engine.domain.dataasset.model.domain.LineageDomain;
import com.sh.data.engine.domain.dataasset.service.LineageService;
import com.sh.data.engine.domain.datadev.offline.config.OfflineDevConfig;
import com.sh.data.engine.domain.datadev.offline.model.domain.*;
import com.sh.data.engine.domain.datadev.offline.model.domain.OfflineDevOperationTaskMonitorDomain.MonitorRecord;
import com.sh.data.engine.domain.datadev.offline.model.domain.OfflineDevOperationTaskflowDetailInfoDomain.TaskFlowSubTask;
import com.sh.data.engine.domain.datadev.offline.model.domain.OfflineDevOperationTaskflowDetailInfoDomain.TaskFlowSubTaskDependency;
import com.sh.data.engine.domain.datadev.offline.model.domain.OfflineDevOperationTodayFinishedDomain.RunRecord;
import com.sh.data.engine.domain.datadev.offline.model.enums.OfflineDevJobStatusEnum;
import com.sh.data.engine.domain.datadev.offline.model.enums.OfflineDevNodeTypeEnum;
import com.sh.data.engine.domain.datadev.offline.model.param.OfflineDevOperationJobSearchParam;
import com.sh.data.engine.domain.datadev.offline.model.param.OfflineDevTaskHistoryJobParam;
import com.sh.data.engine.domain.datadev.offline.model.param.OfflineDevTaskRunParam;
import com.sh.data.engine.domain.datadev.offline.model.param.OfflineDevTaskRunParam.OfflinedevTaskScriptParamParam;
import com.sh.data.engine.domain.datadev.offline.model.param.OfflineDevTaskScriptParam;
import com.sh.data.engine.domain.datadev.offline.runner.*;
import com.sh.data.engine.domain.datadev.offline.service.*;
import com.sh.data.engine.domain.integration.datasource.model.domain.DataSourceDomain;
import com.sh.data.engine.domain.integration.datasource.service.DataSourceService;
import com.sh.data.engine.domain.util.ConvertUtil;
import com.sh.data.engine.domain.util.SqlParserUtil;
import com.sh.data.engine.domain.workspace.manager.model.domain.UserDomain;
import com.sh.data.engine.domain.workspace.manager.service.UserService;
import com.sh.data.engine.infrastructure.config.FileStorageConfiguration;
import com.sh.data.engine.repository.dao.base.EngineBaseModel;
import com.sh.data.engine.repository.dao.datadev.DatadevJobMapper;
import com.sh.data.engine.repository.dao.datadev.entity.*;
import jakarta.servlet.http.HttpServletResponse;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.time.DateFormatUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Lazy;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.io.IOException;
import java.io.OutputStream;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.sql.Timestamp;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.time.format.DateTimeFormatter;
import java.time.temporal.ChronoUnit;
import java.util.*;
import java.util.concurrent.*;
import java.util.stream.Collectors;

@Service
@Slf4j
public class DatadevJobServiceImpl extends ServiceImpl<DatadevJobMapper, DatadevJobEntity>
    implements DatadevJobService {

    @Value("${data.engine.localHostIp:}")
    private String localHostIp;

    @Value("${data.engine.metadataDSType:PostgreSQL}")
    private String metadataDSType;

    @Lazy
    @Autowired
    private DatadevMenuNodeService datadevMenuNodeService;

    @Autowired
    private DatadevVnodeSerivce datadevVnodeSerivce;

    @Lazy
    @Autowired
    private OfflineDevMenuNodeService offlineDevMenuNodeService;

    @Autowired
    private DatadevSubNodeService datadevSubNodeService;

    @Autowired
    private DatadevSubNodeDependencyService datadevSubNodeDependencyService;

    @Autowired
    private LogService logService;

    @Autowired
    private UserService userService;

    @Lazy
    @Autowired
    private DataSourceService dataSourceService;

    @Autowired
    private LineageService lineageService;

    @Autowired
    private DistributedService distributedService;

    @Autowired
    private DatadevScriptService datadevScriptService;

    @Autowired
    private DatadevOfflineSyncService datadevOfflineSyncService;

    @Autowired
    private OfflineDevConfig offlineDevConfig;

    @Autowired
    private FileStorageConfiguration fileStorageConfiguration;

    private ExecutorService executorService;

    private Map<Long, Collection<OfflineDevBaseJobRunner>> jobIdJobRunnerMap = new HashMap<>();

    @PostConstruct
    public void init() {
        ThreadFactory threadFactory =
            new ThreadFactoryBuilder().setNameFormat("offline-dev-pool-%d").build();

        executorService =
            new ThreadPoolExecutor(
                offlineDevConfig.getJobThreadNum(),
                offlineDevConfig.getJobThreadNum(),
                0L,
                TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<>(),
                threadFactory,
                new ThreadPoolExecutor.AbortPolicy());
    }

    @Override
    public DatadevJobEntity findById(Long id) {
        return this.getById(id);
    }

    @Override
    public Integer findLastAutoExecutionRecordByNodeId(Long nodeId, Date planTime) {
        LambdaQueryWrapper<DatadevJobEntity> eq =
            new LambdaQueryWrapper<DatadevJobEntity>()
                .eq(DatadevJobEntity::getNodeId, nodeId)
                .lt(DatadevJobEntity::getPlanTime, planTime)
                .eq(DatadevJobEntity::getExecType, 1)
                .orderByDesc(DatadevJobEntity::getId)
                .eq(DatadevJobEntity::getIsSubTask, 0);
        final DatadevJobEntity devJob = getOne(eq, false);
        if (null == devJob) {
            return null;
        }

        return devJob.getStatus();
    }

    @Override
    public void insertOne(DatadevJobEntity entity) {
        this.save(entity);
    }

    @Override
    public void insertBatch(List<DatadevJobEntity> entityList) {
        if (CollectionUtils.isEmpty(entityList)) {
            return;
        }
        this.saveBatch(entityList);
    }

    @Override
    public boolean haveRecordByNodeId(Long nodeId) {
        LambdaQueryWrapper<DatadevJobEntity> eq =
            new LambdaQueryWrapper<DatadevJobEntity>().eq(DatadevJobEntity::getNodeId, nodeId);
        long count = this.count(eq);
        return count > 0;
    }

    @Override
    public DatadevJobEntity findTaskFlowLastRecord(Long nodeId) {

        LambdaQueryWrapper<DatadevJobEntity> lambdaQueryWrapper =
            new LambdaQueryWrapper<DatadevJobEntity>()
                .eq(DatadevJobEntity::getNodeId, nodeId)
                .orderByDesc(DatadevJobEntity::getPlanTime);

        return getOne(lambdaQueryWrapper, false);
    }

    @Override
    public void updateStatus(Long id, Integer status, boolean setStartTime, boolean setEndTime) {

        DatadevJobEntity datadevJobEntity = getById(id);
        if (setStartTime) {
            datadevJobEntity.setStartTime(new Date());
        }
        if (setEndTime) {
            datadevJobEntity.setEndTime(new Date());
        }

        datadevJobEntity.setStatus(status);
        updateById(datadevJobEntity);

        final Collection<OfflineDevBaseJobRunner> offlineDevBaseJobRunners = jobIdJobRunnerMap.get(id);
        if (CollectionUtils.isNotEmpty(offlineDevBaseJobRunners)) {
            offlineDevBaseJobRunners.forEach(jobRunner -> jobRunner.stopJob());
            jobIdJobRunnerMap.remove(id);
        }
    }

    @Override
    public void updateStatusFinished(
        Long id, Integer status, boolean setStartTime, boolean setEndTime) {

        DatadevJobEntity datadevJobEntity = getById(id);
        if (setStartTime) {
            datadevJobEntity.setStartTime(new Date());
        }
        if (setEndTime) {
            datadevJobEntity.setEndTime(new Date());
        }
        if (status == OfflineDevJobStatusEnum.SUCCESS.getCode()
            || status == OfflineDevJobStatusEnum.CANCEL.getCode()
            || status == OfflineDevJobStatusEnum.FAILURE.getCode()
            || status == OfflineDevJobStatusEnum.PART_FAILED.getCode()) {
            jobIdJobRunnerMap.remove(id);
        }
        datadevJobEntity.setStatus(status);
        updateById(datadevJobEntity);
    }

    @Override
    public Long startTaskFlowJob(Long taskFlowNodeId, int execType, Long projectId, String userId) {
        final DatadevMenuNodeEntity taskFlowNode =
            offlineDevMenuNodeService.findByIdAndProjectId(taskFlowNodeId, projectId);

        List<DatadevSubNodeEntity> subNodeList1 =
            datadevSubNodeService.getDatadevSubNodeList(taskFlowNode.getId());

        List<DatadevSubNodeDependencyEntity> subNodeDependencyEntities =
            datadevSubNodeDependencyService.getDatadevSubNodeDepenDencyList(taskFlowNodeId);

        final List<DatadevSubNodeDependencyEntity> subTaskDependencyList =
            Optional.ofNullable(subNodeDependencyEntities).orElse(new ArrayList<>()).stream()
                .filter(sb -> !Objects.equals(sb.getRowState(), 0))
                .collect(Collectors.toList());

        final List<Long> subTaskNodeIdList =
            subNodeList1.stream()
                .filter(
                    sb ->
                        (Objects.nonNull(sb)
                            && Objects.nonNull(sb.getNodeId())
                            && !Objects.equals(sb.getRowState(), 0)))
                .map(sb -> sb.getNodeId())
                .collect(Collectors.toList());

        final List<DatadevMenuNodeEntity> subNodeList =
            offlineDevMenuNodeService.findByIdList(subTaskNodeIdList);

        DatadevMenuNodeEntity vnode =
            datadevMenuNodeService.getDatadevMenuNodeByFlowId(
                taskFlowNodeId, OfflineDevNodeTypeEnum.VNODE_START);
        Set<Long> waitRunningSubNodeIdSet = new HashSet<>();
        waitRunningSubNodeIdSet.add(vnode.getId());

        //    for (OfflineDevTaskConfigEntity.TaskFlowSubTaskDependency dependency :
        // subTaskDependencyList) {
        //      final String upNodeId = dependency.getUpNodeId();
        //      final String downNodeId = dependency.getDownNodeId();
        //      waitRunningSubNodeIdSet.add(upNodeId);
        //      waitRunningSubNodeIdSet.add(downNodeId);
        //    }

        List<DatadevSubNodeDependencyEntity> temp =
            ConvertUtil.copyProperties(subTaskDependencyList, DatadevSubNodeDependencyEntity.class);

        getReachabledNode(waitRunningSubNodeIdSet, temp);
        subTaskDependencyList.removeAll(temp);
        final List<DatadevMenuNodeEntity> waitRunningSubNodeList =
            subNodeList.stream()
                .filter(sb -> waitRunningSubNodeIdSet.contains(sb.getId()))
                .collect(Collectors.toList());

        Date planTime = new Date();

        if (execType == 1) {
            DatadevVnodeEntity vNodeConfig = datadevVnodeSerivce.getDatadevVnodeByNodeId(vnode.getId());
            // 获取计划开始时间
            planTime = vNodeConfig.getNextExecTime();
            // 计算下一次计划开始时间
            Date nextTime =
                ScheduleCycleUtil.getNextExecutionTime(
                    vNodeConfig.getCycle(),
                    vNodeConfig.getExecPoint(),
                    vNodeConfig.getEffectiveDateStart(),
                    vNodeConfig.getEffectiveDateEnd(),
                    planTime,
                    vNodeConfig.getStartTimeHour(),
                    vNodeConfig.getStartTimeMin(),
                    vNodeConfig.getEndTimeHour(),
                    vNodeConfig.getEndTimeMin());
            vNodeConfig.setNextExecTime(nextTime);
            datadevVnodeSerivce.updateById(vNodeConfig);

            // 看当前计划执行时间任务是否已经有执行记录,若有则不再执行
            LambdaQueryWrapper<DatadevJobEntity> eq =
                new LambdaQueryWrapper<DatadevJobEntity>()
                    .eq(DatadevJobEntity::getPlanTime, planTime)
                    .eq(DatadevJobEntity::getExecType, 1)
                    .eq(DatadevJobEntity::getNodeId, taskFlowNode.getId());

            final List<DatadevJobEntity> jobEntities = list(eq);
            if (CollectionUtils.isNotEmpty(jobEntities)) {
                log.info("当前计划时间已存在执行记录，不再执行，工作流节点id:[{}]", taskFlowNode.getId());
                return null;
            }

            // 判断上个周期是否执行完成
            if (vNodeConfig.getCycleDependency() != null && vNodeConfig.getCycleDependency().equals(1)) {
                // 自依赖等待上一周期结束
                // 离plantime最近的一次执行记录
                final Integer status = findLastAutoExecutionRecordByNodeId(taskFlowNode.getId(), planTime);
                // 如果最近的一次还在执行中 不满足
                if (status != null && status.equals(OfflineDevJobStatusEnum.RUNNING.getCode())) {
                    log.info(
                        "上一周期执行还未结束，等待下一次执行，工作流节点id:[{}],计划时间：[{}]",
                        taskFlowNode.getId(),
                        DateFormatUtils.format(planTime, "yyyy-MM-dd HH:mm:ss"));
                    return null;
                }
            }
        }

        // 开始执行
        String ipInfo = distributedService.getCurrentIpInfo();
        DatadevJobEntity taskFlowJobEntity = new DatadevJobEntity();
        taskFlowJobEntity.setNodeId(taskFlowNodeId);
        taskFlowJobEntity.setProjectId(taskFlowNode.getProjectId());
        taskFlowJobEntity.setStatus(OfflineDevJobStatusEnum.RUNNING.getCode());
        taskFlowJobEntity.setExecType(execType);
        taskFlowJobEntity.setPlanTime(planTime);
        taskFlowJobEntity.setStartTime(new Date());
        taskFlowJobEntity.setNodeType(taskFlowNode.getNodeType());
        taskFlowJobEntity.setIsSubTask(0);
        taskFlowJobEntity.setCreatorId(userId);
        taskFlowJobEntity.setNodeName(taskFlowNode.getName());
        taskFlowJobEntity.setIpAddress(ipInfo);
        HashMap<String, Object> flowJobInfo = Maps.newHashMap();
        flowJobInfo.put("nodes", subNodeList1);
        flowJobInfo.put("edges", subNodeDependencyEntities);
        taskFlowJobEntity.setTaskFlowExecInfo(JSON.toJSONString(flowJobInfo));
        insertOne(taskFlowJobEntity);

        List<DatadevJobEntity> subJobList = new ArrayList<>();

        for (DatadevMenuNodeEntity entity : waitRunningSubNodeList) {
            DatadevJobEntity job = new DatadevJobEntity();
            job.setNodeId(entity.getId());
            job.setProjectId(entity.getProjectId());
            job.setStatus(OfflineDevJobStatusEnum.NO_RUN.getCode());
            job.setExecType(execType);
            job.setIsSubTask(1);
            job.setNodeType(entity.getNodeType());
            job.setTaskFlowJobId(taskFlowJobEntity.getId());
            job.setNodeName(entity.getName());
            job.setCreatorId(userId);
            job.setIpAddress(ipInfo);
            subJobList.add(job);
        }

        insertBatch(subJobList);

        Map<Long, OfflineDevBaseJobRunner> allJobNodeMap =
            constructJobRunner(subJobList, waitRunningSubNodeList, subTaskDependencyList);

        allJobNodeMap.values().forEach(n -> n.setAllJobNodeMap(allJobNodeMap));

        OfflineDevBaseJobRunner rootJobRunner = allJobNodeMap.get(vnode.getId());

        executorService.submit(rootJobRunner);

        jobIdJobRunnerMap.put(taskFlowJobEntity.getId(), allJobNodeMap.values());
//        LogContext.put(LogKey.NAME.getKey(), taskFlowNode.getName());
        return taskFlowJobEntity.getId();
    }

    private void getReachabledNode(
        Set<Long> reachRunnable, List<DatadevSubNodeDependencyEntity> subTaskDependencyList) {

        Iterator<DatadevSubNodeDependencyEntity> iterator = subTaskDependencyList.iterator();
        boolean flag = false;
        while (iterator.hasNext()) {
            DatadevSubNodeDependencyEntity next = iterator.next();
            Long downNodeId = next.getDownNodeId();
            Long upNodeId = next.getUpNodeId();
            if (reachRunnable.contains(upNodeId)) {
                reachRunnable.add(downNodeId);
                flag = true;
                iterator.remove();
            }
        }
        if (flag) {
            getReachabledNode(reachRunnable, subTaskDependencyList);
        }
    }

    @Override
    public Long startTaskJob(OfflineDevTaskRunParam offlineDevTaskRunParam) {
        Long taskNodeId = offlineDevTaskRunParam.getTaskNodeId();
        Long projectId = offlineDevTaskRunParam.getProjectId();
        int execType = offlineDevTaskRunParam.getExecType();
        String userId = offlineDevTaskRunParam.getUserId();

        final DatadevMenuNodeEntity taskNode =
            offlineDevMenuNodeService.findByIdAndProjectId(taskNodeId, projectId);

        DatadevScriptEntity datadevScriptEntity = null;
        if (StringUtils.isNotBlank(offlineDevTaskRunParam.getScript())) {
            datadevScriptEntity = new DatadevScriptEntity();
            datadevScriptEntity.setScript(offlineDevTaskRunParam.getScript());
            datadevScriptEntity.setParam(JSONArray.toJSONString(offlineDevTaskRunParam.getParams()));
            datadevScriptEntity.setNodeId(taskNodeId);
        }
        String ipInfo = distributedService.getCurrentIpInfo();
        DatadevJobEntity job = new DatadevJobEntity();
        job.setNodeId(taskNodeId);
        job.setProjectId(projectId);
        job.setStatus(OfflineDevJobStatusEnum.RUNNING.getCode());
        job.setStartTime(new Date());
        job.setExecType(execType);
        job.setIsSubTask(0);
        job.setCreatorId(userId);
        job.setNodeName(taskNode.getName());
        job.setIpAddress(ipInfo);
        job.setNodeType(taskNode.getNodeType());
        this.save(job);

        //    if (null != offlineDevTaskRunParam
        //        && StringUtils.isNotBlank(offlineDevTaskRunParam.getScript())) {
        //      ScriptTask scriptTask = new ScriptTask();
        //      scriptTask.setScript(offlineDevTaskRunParam.getScript());
        //      List<OfflinedevTaskScriptParamParam> params = offlineDevTaskRunParam.getParams();
        //      List<Param> paramList =
        //          params.stream()
        //              .map(
        //                  e -> {
        //                    Param param = new Param();
        //                    param.setKey(e.getParamName());
        //                    param.setValue(e.getParamValue());
        //                    return param;
        //                  })
        //              .collect(Collectors.toList());
        //      scriptTask.setParamList(paramList);
        //
        //    }

        final OfflineDevBaseJobRunner jobRunner = getJobRunner(job, taskNode, datadevScriptEntity);
        executorService.submit(jobRunner);

        jobIdJobRunnerMap.put(job.getId(), Lists.newArrayList(jobRunner));

        return job.getId();
    }

    @Override
    public void stopJob(Long jobId, int caseTask) {

        DatadevJobEntity jobEntity = getById(jobId);

        String ipAddress = jobEntity.getIpAddress();
        if (StringUtils.isBlank(ipAddress)) {
            return;
        }
        HashMap<String, String> paramMap = Maps.newHashMap();
        paramMap.put("jobId", String.valueOf(jobId));
        boolean canStop = distributedService.stopTaskPost(ipAddress, paramMap, caseTask);
        if (!canStop) {
            updateStatus(jobId, OfflineDevJobStatusEnum.CANCEL.getCode(), false, true);
        }
//        LogContext.put(LogKey.NAME.getKey(), jobEntity.getNodeName());
    }

    @Override
    public void stopAllFlowJob(List<Long> ids) {
        ids.forEach(
            e -> {
                DatadevJobEntity datadevJobEntity = this.getById(e);
                Long nodeId = datadevJobEntity.getNodeId();
                DatadevMenuNodeEntity menuNodeEntity = datadevMenuNodeService.getById(nodeId);
                Integer status = datadevJobEntity.getStatus();
                if (status == OfflineDevJobStatusEnum.RUNNING.getCode()) {
                    this.stopJob(e, menuNodeEntity.getNodeType());
                }
            });
    }

    @Override
    public void stopAllRunningTaskAllProjectId() {
        String ipInfo = distributedService.getCurrentIpInfo();
        LambdaQueryWrapper<DatadevJobEntity> eq =
            new LambdaQueryWrapper<DatadevJobEntity>()
                .eq(DatadevJobEntity::getStatus, OfflineDevJobStatusEnum.RUNNING.getCode())
                .eq(DatadevJobEntity::getIpAddress, ipInfo);

        List<DatadevJobEntity> list = this.list(eq);
        list.forEach(
            e -> {
                e.setStatus(OfflineDevJobStatusEnum.CANCEL.getCode());
                e.setEndTime(new Date());
            });
        this.updateBatchById(list);
    }

    @Override
    public OfflineDevTaskLogDomain getTaskLog(Long jobId, Integer offset, String rsHasShow) {
        int start = 1;
        if (offset != null && !offset.equals(0)) {
            start = offset;
        }
        OfflineDevTaskLogDomain response = new OfflineDevTaskLogDomain();
        response.setLogEnd(false);
        response.setOffset(offset);
        response.setLogContent(Lists.newArrayList());

        // 结果展示过滤
        List<String> hasShowd = Lists.newArrayList();
        if (StringUtils.isNotBlank(rsHasShow)) {
            String[] rss = rsHasShow.split(",");
            hasShowd.addAll(Arrays.asList(rss));
        }
        List<Map<String, Object>> results = Lists.newArrayList();
        DatadevJobEntity jobEntity = findById(jobId);
        if (!Objects.equals(OfflineDevJobStatusEnum.RUNNING.getCode(), jobEntity.getStatus())) {
            response.setLogEnd(true);
        }
        // 仅仅只是读取单个子任务的日志
        List<String> logs;
        final String dir = fileStorageConfiguration.getDir();
        String subTaskJobLogPath =
            String.format(
                "%s/log/%s/%s.log",
                dir, Constants.LogFileExp.OFFLINE_DEV_JOB_LOG_EXP, jobEntity.getId());

        String rsPath =
            String.format(
                "%s/log/%s/%s", dir, Constants.LogFileExp.OFFLINE_DEV_JOB_RS_EXP, jobEntity.getId());
        try {
            logs = logService.read(subTaskJobLogPath, offset);
            results = logService.readResult(rsPath, 0, 1001, hasShowd);
        } catch (IOException e) {
            log.error(e.getMessage());
            throw new BusinessException("读取日志异常");
        }
        if (org.apache.commons.collections.CollectionUtils.isNotEmpty(logs)) {
            response.setOffset(logs.size() + start);
            response.setLogContent(logs);
        }

        response.setResults(results);
        return response;
    }

    @Override
    public OfflineDevOperationSummaryDomain getOfflineDevSummary(
        Long nodeId, Long projectId, Integer execType) {
        QueryWrapper<DatadevJobEntity> queryChangeReasonWrapper =
            new QueryWrapper<DatadevJobEntity>()
                .select("count(1) as statusNum , status as status")
                .eq("project_id", projectId)
                .and(e -> e.eq("ignored", 0).or()
                    .isNull("ignored"))
                .eq("is_sub_task", 0)
                .groupBy("status");

        if (null != execType) {
            queryChangeReasonWrapper.eq("exec_type", execType);
        }
        if (null != nodeId) {
            queryChangeReasonWrapper.eq("node_id", nodeId);
        }
        LambdaQueryWrapper<DatadevJobEntity> eq =
            new LambdaQueryWrapper<DatadevJobEntity>().eq(DatadevJobEntity::getProjectId, projectId);

        long count = count(eq);

        List<Map<String, Object>> maps = listMaps(queryChangeReasonWrapper);
        OfflineDevOperationSummaryDomain offlineDevOperationSummaryDomain =
            new OfflineDevOperationSummaryDomain();
        maps.forEach(
            e -> {
                Object status = e.get("status");
                Object statusnum = e.get("statusnum");
                if (status == null) {
                    status = e.get("STATUS");
                }
                if (statusnum == null) {
                    statusnum = e.get("STATUSNUM");
                }
                if (statusnum == null) {
                    statusnum = e.get("statusNum");
                }
                String statusNum = statusnum.toString();
                status = Integer.valueOf(status.toString());
                if (status.equals(OfflineDevJobStatusEnum.SUCCESS.getCode())) {
                    offlineDevOperationSummaryDomain.setSuccessNum(Long.valueOf(statusNum));
                } else if (status.equals(OfflineDevJobStatusEnum.FAILURE.getCode())
                    || status.equals(OfflineDevJobStatusEnum.PART_FAILED.getCode())) {
                    Long failedNum = offlineDevOperationSummaryDomain.getFailedNum();
                    offlineDevOperationSummaryDomain.setFailedNum(Long.valueOf(failedNum + statusNum));
                }
            });

        // 计算已发布和未发布的

        List<DatadevMenuNodeEntity> allTaskFlowByProjectId =
            datadevMenuNodeService.getAllTaskFlowByProjectId(projectId);
        long unPublish =
            allTaskFlowByProjectId.stream().filter(e -> Objects.isNull(e.getXxlJobId())).count();
        long publish = allTaskFlowByProjectId.size() - unPublish;
        offlineDevOperationSummaryDomain.setUnPublishNum(unPublish);
        offlineDevOperationSummaryDomain.setPublishNum(publish);
        offlineDevOperationSummaryDomain.setAllTask(Long.valueOf(allTaskFlowByProjectId.size()));
        offlineDevOperationSummaryDomain.setAllTaskRecord(count);
        return offlineDevOperationSummaryDomain;
    }

    @Override
    public OfflineDevOperationSummaryDomain getHistorySummary(Long nodeId) {
        OfflineDevOperationSummaryDomain summaryDomain = new OfflineDevOperationSummaryDomain();
        LambdaQueryWrapper<DatadevJobEntity> success =
            new LambdaQueryWrapper<DatadevJobEntity>().eq(DatadevJobEntity::getNodeId, nodeId);
        success.eq(DatadevJobEntity::getStatus, OfflineDevJobStatusEnum.SUCCESS.getCode());
        long count = this.count(success);
        summaryDomain.setSuccessNum(count);

        LambdaQueryWrapper<DatadevJobEntity> failed =
            new LambdaQueryWrapper<DatadevJobEntity>().eq(DatadevJobEntity::getNodeId, nodeId);
        failed.in(
            DatadevJobEntity::getStatus,
            Lists.newArrayList(
                OfflineDevJobStatusEnum.FAILURE.getCode(),
                OfflineDevJobStatusEnum.PART_FAILED.getCode()));
        long failedCount = this.count(failed);
        summaryDomain.setFailedNum(failedCount);
        return summaryDomain;
    }

    @Override
    public PageResult<OfflineDevOperationJobDomain> getOfflineDevOperationJobPage(
        OfflineDevOperationJobSearchParam offlineDevOperationJobSearchParam) {

        Long projectId = offlineDevOperationJobSearchParam.getProjectId();
        Integer pageNum = offlineDevOperationJobSearchParam.getPageNum();
        Integer pageSize = offlineDevOperationJobSearchParam.getPageSize();
        LambdaQueryWrapper<DatadevJobEntity> lambdaQueryWrapper =
            new LambdaQueryWrapper<DatadevJobEntity>()
                .eq(DatadevJobEntity::getProjectId, projectId)
                .eq(DatadevJobEntity::getIsSubTask, 0);
        List<Integer> nodeTypes = Lists.newArrayList();
        if (StringUtils.isNotEmpty(offlineDevOperationJobSearchParam.getNodeType())) {
            nodeTypes =
                Arrays.stream(offlineDevOperationJobSearchParam.getNodeType().split(","))
                    .map(e -> Integer.valueOf(e))
                    .collect(Collectors.toList());

            if (!nodeTypes.contains(OfflineDevNodeTypeEnum.DIR_TASK_FLOW.getNodeType())) {
                nodeTypes.add(OfflineDevNodeTypeEnum.DIR_TASK_FLOW.getNodeType());
                lambdaQueryWrapper.in(DatadevJobEntity::getNodeType, nodeTypes);
            }
        }

        if (StringUtils.isNotBlank(offlineDevOperationJobSearchParam.getNodeName())) {
            lambdaQueryWrapper.like(
                DatadevJobEntity::getNodeName, offlineDevOperationJobSearchParam.getNodeName());
        }
        if (StringUtils.isNotBlank(offlineDevOperationJobSearchParam.getStatus())) {
            List<Integer> statusList =
                Arrays.stream(offlineDevOperationJobSearchParam.getStatus().split(","))
                    .map(e -> Integer.valueOf(e))
                    .collect(Collectors.toList());
            lambdaQueryWrapper.in(DatadevJobEntity::getStatus, statusList);
        }
        if (StringUtils.isNotBlank(offlineDevOperationJobSearchParam.getExecType())) {
            List<Integer> execType =
                Arrays.stream(offlineDevOperationJobSearchParam.getExecType().split(","))
                    .map(e -> Integer.valueOf(e))
                    .collect(Collectors.toList());
            lambdaQueryWrapper.in(DatadevJobEntity::getExecType, execType);
        }

        if (StringUtils.isNotBlank(offlineDevOperationJobSearchParam.getStartTime())) {
            Date parse =
                DateUtil.parse(
                    offlineDevOperationJobSearchParam.getStartTime(), DATE_PATTERN.yyyyMMddHHmmss);
            lambdaQueryWrapper.ge(DatadevJobEntity::getPlanTime, parse);
        }

        if (StringUtils.isNotBlank(offlineDevOperationJobSearchParam.getEndTime())) {
            Date parse =
                DateUtil.parse(
                    offlineDevOperationJobSearchParam.getEndTime(), DATE_PATTERN.yyyyMMddHHmmss);
            lambdaQueryWrapper.le(DatadevJobEntity::getPlanTime, parse);
        }

        if (Objects.nonNull(offlineDevOperationJobSearchParam.getUserId())) {
            lambdaQueryWrapper.eq(
                DatadevJobEntity::getCreatorId, offlineDevOperationJobSearchParam.getUserId());
        }
        if (StringUtils.isNotBlank(offlineDevOperationJobSearchParam.getNodeType())) {

            List<DatadevMenuNodeEntity> allTaskFlowByProjectId =
                datadevMenuNodeService.getAllTaskFlowByProjectId(projectId);
            List<Long> allFlowId = allTaskFlowByProjectId.stream().map(e -> e.getId())
                .collect(Collectors.toList());
            List<Long> flowIds = Lists.newArrayList();
            if (nodeTypes.contains(OfflineDevNodeTypeEnum.DIR_TASK_FLOW.getNodeType())
                && nodeTypes.size() == 1) {
                lambdaQueryWrapper.in(DatadevJobEntity::getNodeId, allFlowId);
            } else {
                for (DatadevMenuNodeEntity datadevMenuNodeEntity : allTaskFlowByProjectId) {
                    Long id = datadevMenuNodeEntity.getId();
                    boolean b = datadevSubNodeService.checkTaskFlowContainTaskType(id, nodeTypes);
                    if (!b) {
                        flowIds.add(id);
                    }
                }
                if (CollectionUtils.isNotEmpty(flowIds)) {
                    lambdaQueryWrapper.notIn(DatadevJobEntity::getNodeId, flowIds);
                }
            }
        }
        String orderField = offlineDevOperationJobSearchParam.getOrderField();
        if (Objects.nonNull(orderField)) {
            if (orderField.equals("planTime")) {
                lambdaQueryWrapper.last(
                    "order by " + " plan_time " + offlineDevOperationJobSearchParam.getOrder());
            }
            if (orderField.equals("startTime")) {
                lambdaQueryWrapper.last(
                    "order by " + " start_time " + offlineDevOperationJobSearchParam.getOrder());
            }
            if (orderField.equals("endTime")) {
                lambdaQueryWrapper.last(
                    "order by " + " end_time " + offlineDevOperationJobSearchParam.getOrder());
            }
            if (orderField.equals("time")) {
                lambdaQueryWrapper.last(
                    "order by "
                        + " (COALESCE(end_time,start_time) - start_time) "
                        + offlineDevOperationJobSearchParam.getOrder());
            }
        } else {
            lambdaQueryWrapper.orderByDesc(DatadevJobEntity::getStartTime);
        }

        IPage<OfflineDevOperationJobDomain> r2 =
            page(new Page<>(pageNum, pageSize), lambdaQueryWrapper)
                .convert(e -> Convert.convert(OfflineDevOperationJobDomain.class, e));

        List<OfflineDevOperationJobDomain> result = r2.getRecords();

        List<Long> nodeIds = result.stream().map(e -> e.getNodeId()).collect(Collectors.toList());

        List<OfflineDevNodeDomain> nodeListByIds =
            offlineDevMenuNodeService.getNodeListByIds(nodeIds, projectId);

        Map<Long, Integer> idMapType = Maps.newHashMap();
        Map<Long, OfflineDevTaskFlowVNodeDetailDomain> map = Maps.newHashMap();
        for (OfflineDevNodeDomain offlineDevNodeDomain : nodeListByIds) {
            // 判断是否是工作流
            if (OfflineDevNodeTypeEnum.DIR_TASK_FLOW
                .getNodeType()
                .equals(offlineDevNodeDomain.getNodeType())) {
                OfflineDevTaskFlowVNodeDetailDomain vNodeDetailDomain =
                    offlineDevMenuNodeService.getNodeDetailByTaskFlowId(
                        offlineDevNodeDomain.getId(), projectId);
                if (vNodeDetailDomain != null) {
                    map.put(offlineDevNodeDomain.getId(), vNodeDetailDomain);
                }
            }
            idMapType.put(offlineDevNodeDomain.getId(), offlineDevNodeDomain.getNodeType());
        }

        result.forEach(
            e -> {
                e.setCanReRun(true);
                if (e.getEndTime() != null && e.getStartTime() != null) {
                    LocalDateTime start =
                        LocalDateTime.ofInstant(e.getStartTime().toInstant(), ZoneId.systemDefault());
                    LocalDateTime end =
                        LocalDateTime.ofInstant(e.getEndTime().toInstant(), ZoneId.systemDefault());
                    final long between = ChronoUnit.SECONDS.between(start, end);
                    e.setTime(between);
                }
                if (checkIsTaskFlow(e.getNodeId())) {
                    e.setHaveChildren(true);
                    OfflineDevTaskFlowVNodeDetailDomain offlineDevTaskFlowVNodeDetailDomain =
                        map.get(e.getNodeId());
                    String schedule = getScheduleInfo(offlineDevTaskFlowVNodeDetailDomain);
                    e.setCycle(offlineDevTaskFlowVNodeDetailDomain.getCycle());
                    e.setScheduleInfo(schedule);
                }
                e.setNodeType(idMapType.get(e.getNodeId()));
                if (Objects.nonNull(e.getCreatorId())) {
                    UserDomain userDomain = userService.getUserById(e.getCreatorId());
                    e.setUserName(userDomain.getUserName());
                }
            });

        return PageResult.convert(r2);
    }

    private String getScheduleInfo(OfflineDevTaskFlowVNodeDetailDomain schedule) {

        StringBuilder sb = new StringBuilder("执行时间: ");
        Integer startTimeHour = schedule.getStartTimeHour();
        Integer startTimeMin = schedule.getStartTimeMin();
        Integer endTimeHour = schedule.getEndTimeHour();
        Integer endTimeMin = schedule.getEndTimeMin();
        String startTimeHourValue = String.valueOf(startTimeHour);
        String startTimeMinValue = String.valueOf(startTimeMin);
        String endTimeHourValue = String.valueOf(endTimeHour);
        String endTimeMinValue = String.valueOf(endTimeMin);

        if (null != startTimeHour && startTimeHour < 10) {
            startTimeHourValue = 0 + startTimeHourValue;
        }
        if (null != startTimeMin && startTimeMin < 10) {
            startTimeMinValue = 0 + startTimeMinValue;
        }
        if (null != endTimeHour && endTimeHour < 10) {
            endTimeHourValue = 0 + endTimeHourValue;
        }
        if (null != endTimeMin && endTimeMin < 10) {
            endTimeMinValue = 0 + endTimeMinValue;
        }
        sb.append(startTimeHourValue);
        sb.append(":");
        sb.append(startTimeMinValue);
        Integer cycle = schedule.getCycle();
        if (cycle == null) {
            return null;
        }
        if (cycle == 0) {
            sb.append("\n执行间隔: ");
            sb.append("每月");
            sb.append(schedule.getExecPoint());
            sb.append("号");
        } else if (cycle == 1) {
            sb.append("\n执行间隔: ");
            sb.append("每周");
            sb.append(schedule.getExecPoint());
        } else if (cycle == 2) {

        } else if (cycle == 3) {
            sb.append("~");
            sb.append(endTimeHourValue);
            sb.append(":");
            sb.append(endTimeMinValue);
            sb.append("\n执行间隔: ");
            sb.append("每");
            sb.append(schedule.getExecPoint());
            sb.append("小时");
        } else if (cycle == 4) {
            sb.append("~");
            sb.append(endTimeHourValue);
            sb.append(":");
            sb.append(endTimeMinValue);
            sb.append("\n执行间隔: ");
            sb.append("每");
            sb.append(schedule.getExecPoint());
            sb.append("分钟");
        }
        return sb.toString();
    }

    private boolean checkIsTaskFlow(Long id) {
        if (Objects.isNull(id)) {
            return false;
        }
        DatadevMenuNodeEntity menuNodeEntity = datadevMenuNodeService.getById(id);
        Integer nodeType = menuNodeEntity.getNodeType();
        if (nodeType.equals(OfflineDevNodeTypeEnum.DIR_TASK_FLOW.getNodeType())) {
            return true;
        }
        return false;
    }

    @Override
    public List<OfflineDevOperationJobDomain> getOfflineDevOperationJobList(
        Long id, Integer nodeType) {

        DatadevJobEntity offlineDevJob = getById(id);

        Long nodeId = offlineDevJob.getNodeId();
        OfflineDevTaskFlowVNodeDetailDomain nodeDetailByTaskFlowId =
            offlineDevMenuNodeService.getNodeDetailByTaskFlowId(nodeId, offlineDevJob.getProjectId());

        LambdaQueryWrapper<DatadevJobEntity> eq =
            new LambdaQueryWrapper<DatadevJobEntity>()
                .eq(DatadevJobEntity::getTaskFlowJobId, id)
                .eq(DatadevJobEntity::getIsSubTask, 1);

        List<DatadevJobEntity> devJobEntityList = list(eq);
        List<OfflineDevOperationJobDomain> list =
            Convert.toList(OfflineDevOperationJobDomain.class, devJobEntityList);

        List<Long> ids = devJobEntityList.stream().map(e -> e.getNodeId()).collect(Collectors.toList());
        Map<Long, Integer> nodeTypeMap = Maps.newHashMap();
        if (CollectionUtils.isNotEmpty(ids)) {
            List<DatadevMenuNodeEntity> byIdList = offlineDevMenuNodeService.findByIdList(ids);
            nodeTypeMap =
                byIdList.stream()
                    .collect(
                        Collectors.toMap(
                            DatadevMenuNodeEntity::getId, DatadevMenuNodeEntity::getNodeType));
        }
        Map<Long, Integer> finalNodeType = nodeTypeMap;
        Iterator<OfflineDevOperationJobDomain> iterator = list.iterator();
        while (iterator.hasNext()) {
            OfflineDevOperationJobDomain next = iterator.next();
            Integer type = finalNodeType.get(next.getNodeId());
            next.setNodeType(type);
            if (Objects.nonNull(nodeType)) {
                if (!type.equals(nodeType)) {
                    iterator.remove();
                }
            }
            next.setCycle(nodeDetailByTaskFlowId.getCycle());
            next.setHaveChildren(false);
            if (next.getEndTime() != null && next.getStartTime() != null) {
                LocalDateTime start =
                    LocalDateTime.ofInstant(next.getStartTime().toInstant(), ZoneId.systemDefault());
                LocalDateTime end =
                    LocalDateTime.ofInstant(next.getEndTime().toInstant(), ZoneId.systemDefault());
                final long between = ChronoUnit.SECONDS.between(start, end);
                next.setTime(between);
            }
            next.setCanReRun(true);
            if (type.equals(OfflineDevNodeTypeEnum.VNODE_START.getNodeType())) {
                next.setCanReRun(false);
            }
            UserDomain userDomain = userService.getUserById(next.getCreatorId());
            next.setUserName(userDomain.getUserName());
            next.setNodeType(type);
        }
        List<OfflineDevOperationJobDomain> rs =
            list.stream()
                .sorted(Comparator.comparing(OfflineDevOperationJobDomain::getNodeId))
                .collect(Collectors.toList());
        return rs;
    }

    @Override
    public Long startTask(Long id, Integer execType, Long projectId, String userId) {
        DatadevJobEntity offlineDevJobEntity = getById(id);
        Long aLong = startTaskByNodeId(offlineDevJobEntity.getNodeId(), projectId, userId, execType);
//        LogContext.put(LogKey.NAME.getKey(), offlineDevJobEntity.getNodeName());
        return aLong;
    }

    @Override
    public Long startTaskByNodeId(Long nodeId, Long projectId, String userId, Integer execType) {
        DatadevMenuNodeEntity menuNode = datadevMenuNodeService.getById(nodeId);
        if (menuNode.getNodeType().equals(OfflineDevNodeTypeEnum.DIR_TASK_FLOW.getNodeType())) {
            Long logId = startTaskFlowJob(nodeId, execType, projectId, userId);
            return logId;
        } else {
            DatadevScriptEntity scriptTaskConfig = datadevScriptService.getDatadevScriptByNodeId(nodeId);

            OfflineDevTaskRunParam offlineDevTaskRunParam = new OfflineDevTaskRunParam();
            offlineDevTaskRunParam.setExecType(execType);
            offlineDevTaskRunParam.setProjectId(projectId);
            offlineDevTaskRunParam.setUserId(userId);
            offlineDevTaskRunParam.setTaskNodeId(menuNode.getId());
            if (null != scriptTaskConfig) {
                String param = scriptTaskConfig.getParam();
                List<Param> paramList = JSONArray.parseArray(param, Param.class);

                List<OfflinedevTaskScriptParamParam> collect = Lists.newArrayList();
                if (CollectionUtils.isNotEmpty(paramList)) {
                    collect =
                        paramList.stream()
                            .map(
                                e -> {
                                    OfflinedevTaskScriptParamParam offlinedevTaskScriptParamParam =
                                        new OfflinedevTaskScriptParamParam();
                                    offlinedevTaskScriptParamParam.setParamName(e.getParamName());
                                    offlinedevTaskScriptParamParam.setParamValue(e.getParamValue());
                                    return offlinedevTaskScriptParamParam;
                                })
                            .collect(Collectors.toList());
                }
                offlineDevTaskRunParam.setScript(scriptTaskConfig.getScript());
                offlineDevTaskRunParam.setParams(collect);
            }
            Long logId = startTaskJob(offlineDevTaskRunParam);
//            LogContext.put(LogKey.NAME.getKey(), menuNode.getName());
            return logId;
        }
    }

    @Override
    public OfflineDevOperationTaskflowDetailInfoDomain getOfflineDevTaskFlowDetailDomain(
        Long flowId) {
        OfflineDevOperationTaskflowDetailInfoDomain offlineDevTaskFlowDetailDomain =
            new OfflineDevOperationTaskflowDetailInfoDomain();

        DatadevMenuNodeEntity menuNodeEntity = datadevMenuNodeService.getById(flowId);

        offlineDevTaskFlowDetailDomain.setName(menuNodeEntity.getName());
        offlineDevTaskFlowDetailDomain.setExecuteType(1);
        offlineDevTaskFlowDetailDomain.setCreateTime(menuNodeEntity.getCreateTime());
        offlineDevTaskFlowDetailDomain.setUpdateTime(menuNodeEntity.getUpdateTime());

        UserDomain userDomain = userService.getUserById(menuNodeEntity.getCreatorId());
        offlineDevTaskFlowDetailDomain.setUserName(userDomain.getUserName());
        DatadevVnodeEntity vnodeStartConfig = datadevVnodeSerivce.getDatadevVnodeByFlowId(flowId);
        if (null != vnodeStartConfig) {
            offlineDevTaskFlowDetailDomain.setCycle(vnodeStartConfig.getCycle());
            offlineDevTaskFlowDetailDomain.setCycleDependency(vnodeStartConfig.getCycleDependency());
        }

        // 获取所有的子任务

        List<DatadevSubNodeEntity> datadevSubNodeList =
            datadevSubNodeService.getDatadevSubNodeList(flowId);
        List<DatadevSubNodeDependencyEntity> datadevSubNodeDepenDencyList =
            datadevSubNodeDependencyService.getDatadevSubNodeDepenDencyList(flowId);

        List<TaskFlowSubTask> taskFlowSubTasks =
            Convert.toList(TaskFlowSubTask.class, datadevSubNodeList);
        List<TaskFlowSubTaskDependency> taskFlowSubTaskDependencies =
            Convert.toList(TaskFlowSubTaskDependency.class, datadevSubNodeDepenDencyList);
        OfflineDevOperationFlowTaskProcessDomain operationFlowTaskProcessDomain =
            new OfflineDevOperationFlowTaskProcessDomain();

        operationFlowTaskProcessDomain.setNodes(taskFlowSubTasks);
        operationFlowTaskProcessDomain.setEdges(taskFlowSubTaskDependencies);

        if (null != operationFlowTaskProcessDomain) {
            List<OfflineDevOperationTaskflowDetailInfoDomain.TaskFlowSubTask> nodes =
                operationFlowTaskProcessDomain.getNodes();
            if (CollectionUtils.isNotEmpty(nodes)) {
                for (OfflineDevOperationTaskflowDetailInfoDomain.TaskFlowSubTask taskFlowSubTask : nodes) {
                    Long nodeId = taskFlowSubTask.getNodeId();
                    if (Objects.isNull(nodeId)) {
                        continue;
                    }
                    DatadevMenuNodeEntity menuNode = offlineDevMenuNodeService.findById(nodeId);
                    taskFlowSubTask.setRowState(menuNode.getRowState());
                    taskFlowSubTask.setNodeName(menuNode.getName());
                    taskFlowSubTask.setNodeType(menuNode.getNodeType());
                }
            }
        }

        List<OfflineDevOperationTaskflowDetailInfoDomain.TaskFlowSubTask> nodes =
            operationFlowTaskProcessDomain.getNodes();
        List<OfflineDevOperationTaskflowDetailInfoDomain.TaskFlowSubTaskDependency> edges =
            operationFlowTaskProcessDomain.getEdges();

        offlineDevTaskFlowDetailDomain.setNodes(nodes);
        offlineDevTaskFlowDetailDomain.setEdges(edges);
        return offlineDevTaskFlowDetailDomain;
    }

    @Override
    public Object getNodeDetail(Long nodeId) throws IOException {

        DatadevMenuNodeEntity menuNode = offlineDevMenuNodeService.findById(nodeId);
        OfflineDevNodeTypeEnum of = OfflineDevNodeTypeEnum.of(menuNode.getNodeType());
        // 离线同步任务
        if (OfflineDevNodeTypeEnum.OFFLINE.equals(of)) {

            DatadevOfflineSyncEntity offlineSyncEntity =
                datadevOfflineSyncService.getDatadevOfflineSyncNodeId(nodeId);

            OfflineDevSyncDomain offlineDevSyncDomain =
                Convert.convert(OfflineDevSyncDomain.class, offlineSyncEntity);

            offlineDevSyncDomain.setNodeType(OfflineDevNodeTypeEnum.OFFLINE.getNodeType());
            return offlineDevSyncDomain;
        }

        if (OfflineDevNodeTypeEnum.VNODE_START.equals(of)) {
            DatadevVnodeEntity vnodeStartConfig =
                datadevVnodeSerivce.getDatadevVnodeByNodeId(menuNode.getId());
            if (null == vnodeStartConfig) {
                return new VnodeStartDomain();
            }
            VnodeStartDomain vnodeStartDomain =
                ConvertUtil.copyProperties(vnodeStartConfig, VnodeStartDomain.class);
            vnodeStartDomain.setNodeType(OfflineDevNodeTypeEnum.VNODE_START.getNodeType());
            return vnodeStartDomain;
        }

        if (OfflineDevNodeTypeEnum.SQL.equals(of)) {
            DatadevScriptEntity scriptTaskConfig =
                datadevScriptService.getDatadevScriptByNodeId(menuNode.getId());
            HashMap<String, Object> res = Maps.newHashMap();
            if (scriptTaskConfig == null) {
                List<OfflineDevOperationLineageDomain> result = Lists.newArrayList();
                res.put("offlineData", result);
                res.put("nodeType", OfflineDevNodeTypeEnum.SQL.getNodeType());
                return result;
            }
            Long projectId = menuNode.getProjectId();
            DataSourceDomain systemDataSourceByProjectId =
                dataSourceService.getSystemDataSourceByProjectId(projectId);
            String dsType = systemDataSourceByProjectId.getDsType();
            Database from = Database.from(dsType);
            if (dsType.equalsIgnoreCase("sap")) {
                dsType = "sapdb";
            }
            DbType dbType = DbType.of(dsType.toLowerCase());
            String script = scriptTaskConfig.getScript();
            List<LineageDomain> allLineage = Lists.newArrayList();
            List<String> scripts = Lists.newArrayList();

            if (dsType.equals(DSType.Sap.name())) {
                scripts = SqlParserUtil.split(script, DbType.sapdb);
                Map<String, String> map = null;
                for (String sql : scripts) {
                    map = SqlParserUtil.getConsanguinityForSap(sql);
                    String source = map.get("source");
                    String target = map.get("target");
                    if (StringUtils.isNotBlank(source) && StringUtils.isNotBlank(target)) {
                        String[] split = source.split(",");
                        for (String str : split) {
                            LineageDomain lineageDomain = new LineageDomain();
                            lineageDomain.setSourceTable(str);
                            lineageDomain.setTargetTable(target);
                            allLineage.add(lineageDomain);
                        }
                    }
                }
            } else {
                scripts = SqlParserUtil.split(script, dbType);

                for (String sql : scripts) {
                    List<LineageDomain> lineageDomains =
                        lineageService.analyseSqlLineage(sql, from, projectId);
                    allLineage.addAll(lineageDomains);
                }
            }

            List<OfflineDevOperationLineageDomain> result =
                ConvertUtil.copyProperties(allLineage, OfflineDevOperationLineageDomain.class);

            res.put("offlineData", result);
            res.put("nodeType", OfflineDevNodeTypeEnum.SQL.getNodeType());
            return res;
        }

        //    if (OfflineDevNodeTypeEnum.DATAQUALITY.equals(of)) {
        //      DataqualityTask dataqualityTaskConfig = taskConfig.getDataqualityTaskConfig();
        //
        //      OfflineDevOpearationQualityTemplateInfoDomain result =
        //          ConvertUtil.copyProperties(
        //              dataqualityTaskConfig, OfflineDevOpearationQualityTemplateInfoDomain.class);
        //
        //      // todo 等待数据质量初始化
        //      return result;
        //    }
        return null;
    }

    @Override
    public void downloadTaskResult(
        Long jobId, String result, Long projectId, HttpServletResponse response) {

        String dir = fileStorageConfiguration.getDir();
        String subTaskJobLogPath =
            String.format("%s/log/%s/%s", dir, Constants.LogFileExp.OFFLINE_DEV_JOB_RS_EXP, jobId);
        List<List<String>> rss = logService.readHiveRs(subTaskJobLogPath, 1, result);
        if (org.apache.commons.collections.CollectionUtils.isNotEmpty(rss)) {
            ExcelWriter writer = null;
            OutputStream outputStream = null;
            try {

                outputStream = response.getOutputStream();
                // 添加响应头信息
                response.setHeader(
                    "Content-disposition", "attachment; filename=" + UUID.randomUUID() + ".xlsx");
                response.setContentType("application/msexcel;charset=UTF-8"); // 设置类型
                response.setHeader("Pragma", "No-cache"); // 设置头
                response.setHeader("Cache-Control", "no-cache"); // 设置头
                response.setDateHeader("Expires", 0); // 设置日期头

                writer = new ExcelWriter(outputStream, ExcelTypeEnum.XLSX, false);

                WriteSheet sheet = EasyExcel.writerSheet("sheet").build();

                // 获取数据
                writer.write(rss, sheet);
                writer.finish();
                outputStream.flush();
            } catch (IOException e) {
                log.error(e.getMessage(), e);
            } finally {
                try {
                    response.getOutputStream().close();
                } catch (IOException e) {
                    log.error(e.getMessage(), e);
                }
                rss.clear();
            }
        }
    }

    @Override
    public PageResult<OfflineDevOperationJobDomain> getOfflineDevOperationHistoryJobPage(
        OfflineDevTaskHistoryJobParam param) {

        LambdaQueryWrapper<DatadevJobEntity> eq =
            new LambdaQueryWrapper<DatadevJobEntity>()
                .eq(DatadevJobEntity::getNodeId, param.getNodeId())
                .eq(DatadevJobEntity::getProjectId, param.getProjectId());
        if (null != param.getExecType()) {
            eq.eq(DatadevJobEntity::getExecType, param.getExecType());
        }

        if (null != param.getTaskStatus()) {
            eq.eq(DatadevJobEntity::getStatus, param.getTaskStatus());
        }

        String orderField = param.getOrderField();
        if (Objects.nonNull(orderField)) {
            if (orderField.equals("startTime")) {
                eq.last("order by " + " start_time " + param.getOrder());
            }
            if (orderField.equals("endTime")) {
                eq.last("order by " + " end_time " + param.getOrder());
            }
            if (orderField.equals("time")) {
                eq.last("order by " + " (COALESCE(end_time,start_time) - start_time) " + param.getOrder());
            }
        } else {
            eq.orderByDesc(DatadevJobEntity::getStartTime);
        }

        IPage<OfflineDevOperationJobDomain> pageInfo =
            page(new Page<>(param.getPageNum(), param.getPageSize()), eq)
                .convert(e -> Convert.convert(OfflineDevOperationJobDomain.class, e));

        List<OfflineDevOperationJobDomain> result = pageInfo.getRecords();

        DatadevMenuNodeEntity offlineDevMenuNodeEntity =
            datadevMenuNodeService.getById(param.getNodeId());

        Integer nodeType = offlineDevMenuNodeEntity.getNodeType();
        OfflineDevTaskFlowVNodeDetailDomain taskFlowVNodeDetail = null;
        if (OfflineDevNodeTypeEnum.DIR_TASK_FLOW.getNodeType().equals(nodeType)) {
            DatadevMenuNodeEntity taskFlowVnodeByTaskFlowId =
                offlineDevMenuNodeService.getTaskFlowVnodeByTaskFlowId(
                    param.getNodeId(), param.getProjectId());

            taskFlowVNodeDetail =
                offlineDevMenuNodeService.getTaskFlowVNodeDetail(
                    taskFlowVnodeByTaskFlowId.getId(), param.getProjectId());
        }

        OfflineDevTaskFlowVNodeDetailDomain finalTaskFlowVNodeDetail = taskFlowVNodeDetail;
        result.forEach(
            e -> {
                e.setCanReRun(false);
                if (e.getEndTime() != null && e.getStartTime() != null) {
                    LocalDateTime start =
                        LocalDateTime.ofInstant(e.getStartTime().toInstant(), ZoneId.systemDefault());
                    LocalDateTime end =
                        LocalDateTime.ofInstant(e.getEndTime().toInstant(), ZoneId.systemDefault());
                    final long between = ChronoUnit.SECONDS.between(start, end);
                    e.setTime(between);
                }
                UserDomain userDomain = userService.getUserById(e.getCreatorId());
                if (null != userDomain) {
                    e.setUserName(userDomain.getUserName());
                }
                if (null != finalTaskFlowVNodeDetail) {
                    e.setCycle(finalTaskFlowVNodeDetail.getCycle());
                }
                if (nodeType.equals(OfflineDevNodeTypeEnum.DIR_TASK_FLOW.getNodeType())) {
                    e.setHaveChildren(true);
                }
            });
        return PageResult.convert(pageInfo);
    }

    @Override
    public List<OfflineDevTaskFlowStatusDomain> getTaskFlowStatus(Long jobId) {

        LambdaQueryWrapper<DatadevJobEntity> eq =
            new LambdaQueryWrapper<DatadevJobEntity>().eq(DatadevJobEntity::getTaskFlowJobId, jobId);

        List<DatadevJobEntity> jobEntity = list(eq);

        final DatadevJobEntity taskFlowJob = getById(jobId);
        final Integer taskFlowJobStatus = taskFlowJob.getStatus();

        List<OfflineDevTaskFlowStatusDomain> domain =
            jobEntity.stream()
                .map(
                    e -> {
                        OfflineDevTaskFlowStatusDomain taskFlowStatusDomain =
                            new OfflineDevTaskFlowStatusDomain();

                        Integer jobStatus = e.getStatus();
                        // 如果工作流是停止状态，将子节点中是运行中的状态改为停止状态，解决连线问题
                        if (Objects.equals(taskFlowJobStatus, OfflineDevJobStatusEnum.CANCEL.getCode())
                            && Objects.equals(jobStatus, OfflineDevJobStatusEnum.RUNNING.getCode())) {
                            jobStatus = OfflineDevJobStatusEnum.CANCEL.getCode();
                        }

                        taskFlowStatusDomain.setTaskStatus(jobStatus);
                        taskFlowStatusDomain.setNodeId(e.getNodeId());
                        return taskFlowStatusDomain;
                    })
                .collect(Collectors.toList());

        return domain;
    }

    @Override
    public Long startOfflineApiTask(Long nodeId, int execType, Long projectId, String userId) {

        final DatadevMenuNodeEntity taskNode =
            offlineDevMenuNodeService.findByIdAndProjectId(nodeId, projectId);

        String ipInfo = getIpInfoByLogId();
        String localHostIp = this.localHostIp;
        if (StringUtils.isNotBlank(localHostIp)) {
            ipInfo = localHostIp;
        }

        DatadevJobEntity job = new DatadevJobEntity();
        job.setNodeId(nodeId);
        job.setProjectId(projectId);
        job.setStatus(OfflineDevJobStatusEnum.RUNNING.getCode());
        job.setStartTime(new Date());
        job.setExecType(execType);
        job.setIsSubTask(0);
        job.setCreatorId(userId);
        job.setNodeType(taskNode.getNodeType());
        job.setNodeName(taskNode.getName());
        job.setIpAddress(ipInfo);
        insertOne(job);

        final OfflineDevBaseJobRunner jobRunner = getJobRunner(job, taskNode);
        executorService.submit(jobRunner);

        jobIdJobRunnerMap.put(job.getId(), Lists.newArrayList(jobRunner));
//        LogContext.put(LogKey.NAME.getKey(), job.getNodeName());
        return job.getId();
    }

    @Override
    public Long startOfflineSyncTask(Long nodeId, int execType, Long projectId, String userId) {

        final DatadevMenuNodeEntity taskNode =
            offlineDevMenuNodeService.findByIdAndProjectId(nodeId, projectId);

        String ipInfo = distributedService.getCurrentIpInfo();
        DatadevJobEntity job = new DatadevJobEntity();
        job.setNodeId(nodeId);
        job.setProjectId(projectId);
        job.setStatus(OfflineDevJobStatusEnum.RUNNING.getCode());
        job.setStartTime(new Date());
        job.setExecType(execType);
        job.setIsSubTask(0);
        job.setCreatorId(userId);
        job.setNodeType(taskNode.getNodeType());
        job.setNodeName(taskNode.getName());
        job.setIpAddress(ipInfo);
        insertOne(job);

        final OfflineDevBaseJobRunner jobRunner = getJobRunner(job, taskNode);
        executorService.submit(jobRunner);

        jobIdJobRunnerMap.put(job.getId(), Lists.newArrayList(jobRunner));
//        LogContext.put(LogKey.NAME.getKey(), job.getNodeName());
        return job.getId();
    }

    private String getIpInfoByLogId() {
        try {
            InetAddress localHost = InetAddress.getLocalHost();
            String hostAddress = localHost.getHostAddress();
            return hostAddress;
        } catch (UnknownHostException e) {
            log.error("获取本地ip失败", e.getMessage());
        }
        return "localhost";
    }

    @Override
    public Long runRule(Long nodeId, int execType, Long projectId, String userId) {

        final DatadevMenuNodeEntity taskNode =
            offlineDevMenuNodeService.findByIdAndProjectId(nodeId, projectId);

        String ipInfo = getIpInfoByLogId();
        String localHostIp = this.localHostIp;
        if (StringUtils.isNotBlank(localHostIp)) {
            ipInfo = localHostIp;
        }
        DatadevJobEntity job = new DatadevJobEntity();
        job.setNodeId(nodeId);
        job.setProjectId(projectId);
        job.setStatus(OfflineDevJobStatusEnum.RUNNING.getCode());
        job.setStartTime(new Date());
        job.setExecType(execType);
        job.setIsSubTask(0);
        job.setCreatorId(userId);
        job.setNodeType(taskNode.getNodeType());
        job.setNodeName(taskNode.getName());
        job.setIpAddress(ipInfo);
        insertOne(job);

        final OfflineDevBaseJobRunner jobRunner = getJobRunner(job, taskNode);
        executorService.submit(jobRunner);

        jobIdJobRunnerMap.put(job.getId(), Lists.newArrayList(jobRunner));

        return job.getId();
    }

    @Override
    public void dropRecordsByNodeId(Long id) {
        LambdaQueryWrapper<DatadevJobEntity> eq =
            new LambdaQueryWrapper<DatadevJobEntity>().eq(DatadevJobEntity::getNodeId, id);
        remove(eq);
    }

    @Override
    public void cleanMonthAgoRecord(Date time) {
        LambdaQueryWrapper<DatadevJobEntity> lt =
            new LambdaQueryWrapper<DatadevJobEntity>().lt(DatadevJobEntity::getCreateTime, time);
        this.remove(lt);
    }

    @Override
    public OfflineDevOperationTodayFinishedDomain getTodayFinished(Long projectId) {

        OfflineDevOperationTodayFinishedDomain todayFinishedDomain =
            new OfflineDevOperationTodayFinishedDomain();
        QueryWrapper<DatadevJobEntity> toDayWrapper =
            new QueryWrapper<DatadevJobEntity>()
                .select("count(1) as count , extract(hour from end_time)  AS endTime")
                .eq("project_id", projectId)
                .ge("end_time", DateUtil.getTodayStart(new Date()))
                .le("end_time", DateUtil.getTodayEnd(new Date()))
                .eq("status", OfflineDevJobStatusEnum.SUCCESS.getCode())
                .eq("exec_type", 1)
                .eq("is_sub_task", 0)
                .groupBy("end_time");
        List<Map<String, Object>> toDayMaps = this.listMaps(toDayWrapper);

        QueryWrapper<DatadevJobEntity> yesterDayWrapper =
            new QueryWrapper<DatadevJobEntity>()
                .select("count(1) as count , extract(hour from end_time)  AS endTime")
                .eq("project_id", projectId)
                .ge("end_time", DateUtil.getSomeDayStart(new Date(), -1))
                .le("end_time", DateUtil.getSomeDayEnd(new Date(), -1))
                .eq("status", OfflineDevJobStatusEnum.SUCCESS.getCode())
                .eq("exec_type", 1)
                .eq("is_sub_task", 0)
                .groupBy("end_time");
        List<Map<String, Object>> yesterdayMaps = this.listMaps(yesterDayWrapper);
        //
        QueryWrapper<DatadevJobEntity> historyWrapper =
            new QueryWrapper<DatadevJobEntity>()
                .select("count(1) as count , extract(hour from end_time) AS endTime")
                .eq("project_id", projectId)
                .le("end_time", DateUtil.getTodayEnd(new Date()))
                .eq("status", OfflineDevJobStatusEnum.SUCCESS.getCode())
                .eq("exec_type", 1)
                .eq("is_sub_task", 0)
                .groupBy("end_time");
        List<Map<String, Object>> historyMaps = this.listMaps(historyWrapper);

        List<RunRecord> todayRecord = Lists.newArrayList();
        toDayMaps.forEach(
            e -> {
                Object count = e.get("count");
                Object date = e.get("endtime");
                if (count == null) {
                    count = e.get("COUNT");
                }
                if (date == null) {
                    date = e.get("ENDTIME");
                }
                if (date == null) {
                    date = e.get("endTime");
                }
                RunRecord runRecord = new RunRecord();
                runRecord.setTime(Double.valueOf(date.toString()).intValue());
                runRecord.setCount(Double.valueOf(count.toString()).longValue());
                todayRecord.add(runRecord);
            });

        List<RunRecord> yesterdayRecord = Lists.newArrayList();
        yesterdayMaps.forEach(
            e -> {
                Object count = e.get("count");
                Object date = e.get("endtime");
                if (count == null) {
                    count = e.get("COUNT");
                }
                if (date == null) {
                    date = e.get("ENDTIME");
                }
                if (date == null) {
                    date = e.get("endTime");
                }
                RunRecord runRecord = new RunRecord();
                runRecord.setTime(Double.valueOf(date.toString()).intValue());
                runRecord.setCount(Double.valueOf(count.toString()).longValue());
                yesterdayRecord.add(runRecord);
            });

        LambdaQueryWrapper<DatadevJobEntity> large =
            new LambdaQueryWrapper<DatadevJobEntity>()
                .eq(DatadevJobEntity::getIsSubTask, 0)
                .eq(DatadevJobEntity::getExecType, 1)
                .last("order by id desc limit 1")
                .eq(DatadevJobEntity::getProjectId, projectId);

        LambdaQueryWrapper<DatadevJobEntity> small =
            new LambdaQueryWrapper<DatadevJobEntity>()
                .eq(DatadevJobEntity::getIsSubTask, 0)
                .eq(DatadevJobEntity::getExecType, 1)
                .last("order by id asc limit 1")
                .eq(DatadevJobEntity::getProjectId, projectId);

        DatadevJobEntity largeOne = this.getOne(large);
        DatadevJobEntity smallOne = this.getOne(small);
        List<RunRecord> historyRecord = Lists.newArrayList();
        if (Objects.nonNull(largeOne) && Objects.nonNull(smallOne)) {
            Date largeTime = largeOne.getStartTime();
            Date smallTime = smallOne.getStartTime();
            if (Objects.isNull(largeTime)) {
                largeTime = new Date();
            }
            if (Objects.isNull(smallTime)) {
                smallTime = new Date();
            }
            LocalDateTime start = LocalDateTime.ofInstant(smallTime.toInstant(), ZoneId.systemDefault());
            LocalDateTime end = LocalDateTime.ofInstant(largeTime.toInstant(), ZoneId.systemDefault());
            long daysDiff = ChronoUnit.DAYS.between(start, end);
            historyMaps.forEach(
                e -> {
                    Object count = e.get("count");
                    Object date = e.get("endtime");
                    if (count == null) {
                        count = e.get("COUNT");
                    }
                    if (date == null) {
                        date = e.get("ENDTIME");
                    }
                    if (date == null) {
                        date = e.get("endTime");
                    }
                    RunRecord runRecord = new RunRecord();
                    runRecord.setTime(Double.valueOf(date.toString()).intValue());
                    count = Double.valueOf(count.toString()).longValue();
                    long result = Math.round((Long) count / (daysDiff + 1));
                    runRecord.setCount(result);
                    historyRecord.add(runRecord);
                });
        }

        todayFinishedDomain.setTodayRecord(changeData(todayRecord));
        todayFinishedDomain.setYesterdayRecord(changeData(yesterdayRecord));
        todayFinishedDomain.setHistoryAverageRecord(changeData(historyRecord));

        return todayFinishedDomain;
    }

    private List<RunRecord> changeData(List<RunRecord> todayRecord) {
        List<Integer> nums =
            Lists.newArrayList(
                0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23);

        List<Integer> r1 = todayRecord.stream().map(e -> e.getTime()).collect(Collectors.toList());
        nums.removeAll(r1);
        if (CollectionUtils.isNotEmpty(r1)) {
            List<RunRecord> finalTodayRecord = todayRecord;
            nums.forEach(
                e -> {
                    RunRecord runRecord = new RunRecord();
                    runRecord.setTime(e);
                    runRecord.setCount(0L);
                    finalTodayRecord.add(runRecord);
                });
        }
        List<RunRecord> result =
            todayRecord.stream()
                .sorted(Comparator.comparing(RunRecord::getTime))
                .collect(Collectors.toList());
        return result;
    }

    @Override
    public List<OfflineDevOperationFailedTopDomain> getFailedTaskTop(
        Long projectId, Integer timeType) {
        Date date = new Date();
        Calendar calendar = Calendar.getInstance();
        calendar.setTime(date);
        if (timeType == 0) {
            calendar.add(Calendar.DAY_OF_MONTH, -3);
        } else if (timeType == 1) {
            calendar.add(Calendar.DAY_OF_MONTH, -7);
        } else if (timeType == 2) {
            calendar.add(Calendar.MONTH, -1);
        } else if (timeType == 3) {
            calendar.add(Calendar.DAY_OF_MONTH, -1);
        }
        Date time = calendar.getTime();
        List<OfflineDevOperationFailedTopDomain> res = Lists.newArrayList();
        QueryWrapper<DatadevJobEntity> queryChangeReasonWrapper =
            new QueryWrapper<DatadevJobEntity>()
                .select(
                    "node_name as taskName,max(start_time) as lastRunningDate,count(1) as failedNum,node_id as nodeId")
                .eq("project_id", projectId)
                .in(
                    "status",
                    Lists.newArrayList(
                        OfflineDevJobStatusEnum.FAILURE.getCode(),
                        OfflineDevJobStatusEnum.PART_FAILED.getCode()))
                .eq("exec_type", 1)
                .eq("is_sub_task", 0)
                .gt("end_time", time)
                .groupBy("node_id", "node_name")
                .orderByDesc("failedNum")
                .last("limit 10");

        List<Map<String, Object>> maps = this.listMaps(queryChangeReasonWrapper);
        maps.forEach(
            e -> {
                Object taskName = e.get("taskname");
                Object lastRunningDate = e.get("lastrunningdate");
                Object failedNum = e.get("failednum");
                Object nodeId = e.get("nodeid");
                if (taskName == null) {
                    taskName = e.get("TASKNAME");
                }
                if (taskName == null) {
                    taskName = e.get("taskName");
                }
                if (lastRunningDate == null) {
                    lastRunningDate = e.get("LASTRUNNINGDATE");
                }
                if (lastRunningDate == null) {
                    lastRunningDate = e.get("lastRunningDate");
                }
                if (failedNum == null) {
                    failedNum = e.get("FAILEDNUM");
                }
                if (failedNum == null) {
                    failedNum = e.get("failedNum");
                }
                if (nodeId == null) {
                    nodeId = e.get("NODEID");
                }
                if (nodeId == null) {
                    nodeId = e.get("nodeId");
                }
                OfflineDevOperationFailedTopDomain offlineDevOperationFailedTopDomain =
                    new OfflineDevOperationFailedTopDomain();
                offlineDevOperationFailedTopDomain.setFailedNum(Long.valueOf(failedNum.toString()));
                offlineDevOperationFailedTopDomain.setTaskName(String.valueOf(taskName));
                Timestamp timestamp = (Timestamp) lastRunningDate;
                LocalDateTime localDateTime = timestamp.toLocalDateTime();
                offlineDevOperationFailedTopDomain.setLastRunningDate(
                    Date.from(localDateTime.atZone(ZoneId.systemDefault()).toInstant()));
                offlineDevOperationFailedTopDomain.setNodeId(Long.valueOf(nodeId.toString()));
                res.add(offlineDevOperationFailedTopDomain);
            });

        return res;
    }

    @Override
    public List<OfflineDevOperationRunningTimeTopDomain> getRunningTimeTop(
        Long projectId, Date runDate) {

        Date someDayEnd = DateUtil.getSomeDayStart(runDate, 1);

        List<OfflineDevOperationRunningTimeTopDomain> res = Lists.newArrayList();
        String metadataDSType = this.metadataDSType;
        String selectSql =
            "extract(epoch from age(end_time,start_time)) as runningTime ,node_id as nodeId, node_name as taskName,start_time as lastRunningDate,status as taskStatus";
        if (metadataDSType.equals(DSType.DM.name())) {
            selectSql =
                "TIMESTAMPDIFF(SECOND, start_time, end_time) as runningTime , node_id as nodeId,node_name as taskName,start_time as lastRunningDate,status as taskStatus";
        } else if (metadataDSType.equals(DSType.MySQL.name())) {
            selectSql = "TIMESTAMPDIFF(SECOND, start_time, end_time) as runningTime , node_id as nodeId,node_name as taskName,start_time as lastRunningDate,status as taskStatus";
        }

        QueryWrapper<DatadevJobEntity> queryChangeReasonWrapper =
            new QueryWrapper<DatadevJobEntity>()
                .select(selectSql)
                .eq("project_id", projectId)
                .eq("exec_type", 1)
                .eq("is_sub_task", 0)
                .isNotNull("end_time")
                .in(
                    "status",
                    Lists.newArrayList(
                        OfflineDevJobStatusEnum.SUCCESS.getCode(),
                        OfflineDevJobStatusEnum.FAILURE.getCode(),
                        OfflineDevJobStatusEnum.CANCEL.getCode(),
                        OfflineDevJobStatusEnum.PART_FAILED.getCode()))
                .orderByDesc("runningTime")
                .ge("start_time", runDate)
                .le("start_time", someDayEnd)
                .last("limit 10");
        List<Map<String, Object>> maps = this.listMaps(queryChangeReasonWrapper);
        maps.forEach(
            e -> {
                Object runningTime = e.get("runningtime");
                Object taskName = e.get("taskname");
                Object lastRunningDate = e.get("lastrunningdate");
                Object taskStatus = e.get("taskstatus");
                Object nodeId = e.get("nodeid");
                if (taskName == null) {
                    taskName = e.get("TASKNAME");
                }
                if (taskName == null) {
                    taskName = e.get("taskName");
                }
                if (lastRunningDate == null) {
                    lastRunningDate = e.get("LASTRUNNINGDATE");
                }
                if (lastRunningDate == null) {
                    lastRunningDate = e.get("lastRunningDate");
                }
                if (runningTime == null) {
                    runningTime = e.get("RUNNINGTIME");
                }
                if (runningTime == null) {
                    runningTime = e.get("runningTime");
                }
                if (taskStatus == null) {
                    taskStatus = e.get("TASKSTATUS");
                }
                if (taskStatus == null) {
                    taskStatus = e.get("taskStatus");
                }
                if (nodeId == null) {
                    nodeId = e.get("NODEID");
                }
                if (nodeId == null) {
                    nodeId = e.get("nodeId");
                }
                OfflineDevOperationRunningTimeTopDomain timeTopDomain =
                    new OfflineDevOperationRunningTimeTopDomain();
                if (Objects.nonNull(runningTime)) {
                    timeTopDomain.setRunningTime(Double.valueOf(runningTime.toString()));
                }
                if (Objects.nonNull(lastRunningDate)) {
                    Timestamp lastRunningTimeStamp = (Timestamp) lastRunningDate;
                    LocalDateTime localDateTime = lastRunningTimeStamp.toLocalDateTime();
                    timeTopDomain.setLastRunningDate(
                        (Date.from(localDateTime.atZone(ZoneId.systemDefault()).toInstant())));
                }
                timeTopDomain.setTaskName(taskName.toString());
                timeTopDomain.setTaskStatus(Integer.valueOf(String.valueOf(taskStatus)));
                timeTopDomain.setNodeId(Long.valueOf(nodeId.toString()));
                res.add(timeTopDomain);
            });
        return res;
    }

    @Override
    public OfflineDevOperationTaskMonitorDomain getTaskMonitorError(Long projectId) {

        List<MonitorRecord> weekRecord = Lists.newArrayList();
        List<MonitorRecord> monthRecord = Lists.newArrayList();

        String groupBy = "time";
        String metadataDSType = this.metadataDSType;
        String selectSql = "date_trunc('day', end_time) as time,count(1) as totalNum";
        if (metadataDSType.equals(DSType.DM.name())) {
            selectSql = "TRUNC(end_time,'DAY') as time,count(1) as totalNum";
            groupBy = "TRUNC(end_time,'DAY')";
        } else if (metadataDSType.equals(DSType.MySQL.name())) {
            selectSql = "DATE_FORMAT(end_time, '%Y-%m-%d') AS time,COUNT(1) AS totalNum";
        }

        QueryWrapper<DatadevJobEntity> weekQueryWrapper =
            new QueryWrapper<DatadevJobEntity>()
                .select(selectSql)
                .eq("exec_type", 1)
                .eq("is_sub_task", 0)
                .eq("project_id", projectId)
                .in(
                    "status",
                    Lists.newArrayList(
                        OfflineDevJobStatusEnum.FAILURE.getCode(),
                        OfflineDevJobStatusEnum.PART_FAILED.getCode()))
                .ge("end_time", DateUtil.getSomeDayStart(new Date(), -7))
                .groupBy(groupBy)
                .orderByAsc("time");

        QueryWrapper<DatadevJobEntity> monthQueryWrapper =
            new QueryWrapper<DatadevJobEntity>()
                .select(selectSql)
                .eq("exec_type", 1)
                .eq("is_sub_task", 0)
                .eq("project_id", projectId)
                .in(
                    "status",
                    Lists.newArrayList(
                        OfflineDevJobStatusEnum.FAILURE.getCode(),
                        OfflineDevJobStatusEnum.PART_FAILED.getCode()))
                .ge("end_time", DateUtil.getSomeMonthStart(new Date(), -1))
                .groupBy(groupBy)
                .orderByAsc("time");

        List<Map<String, Object>> maps = this.listMaps(weekQueryWrapper);
        maps.forEach(
            e -> {
                Object time = e.get("time");
                Object totalNum = e.get("totalnum");
                if (time == null) {
                    time = e.get("TIME");
                }
                if (totalNum == null) {
                    totalNum = e.get("TOTALNUM");
                }
                if (totalNum == null) {
                    totalNum = e.get("totalNum");
                }
                MonitorRecord runRecord = MonitorRecord.builder().build();
                runRecord.setCount(Long.valueOf(totalNum.toString()));
                runRecord.setTime(time.toString());
                weekRecord.add(runRecord);
            });
        List<Map<String, Object>> maps2 = this.listMaps(monthQueryWrapper);
        maps2.forEach(
            e -> {
                Object time = e.get("time");
                Object totalNum = e.get("totalnum");
                if (time == null) {
                    time = e.get("TIME");
                }
                if (totalNum == null) {
                    totalNum = e.get("TOTALNUM");
                }
                if (totalNum == null) {
                    totalNum = e.get("totalNum");
                }
                MonitorRecord runRecord = MonitorRecord.builder().build();
                runRecord.setCount(Long.valueOf(totalNum.toString()));
                runRecord.setTime(time.toString());
                monthRecord.add(runRecord);
            });

        DateTimeFormatter dateTimeFormatter = DateTimeFormatter.ofPattern("yyyy-MM-dd");
        String endDate = DateUtil.format(new Date(), DATE_PATTERN.yyyyMMdd);
        Date someDayStart = DateUtil.getSomeDayStart(new Date(), -7);
        String startDate = DateUtil.format(someDayStart, DATE_PATTERN.yyyyMMdd);
        LocalDate start = LocalDate.parse(startDate, dateTimeFormatter);
        LocalDate end = LocalDate.parse(endDate, dateTimeFormatter);
        long daysDiff = ChronoUnit.DAYS.between(start, end);
        List<String> dates = weekRecord.stream().map(e -> e.getTime()).collect(Collectors.toList());
        for (int i = 0; i <= daysDiff; i++) {
            String d = start.plusDays(i).format(dateTimeFormatter);
            if (!dates.contains(d)) {
                weekRecord.add(MonitorRecord.builder().time(d).count(0L).build());
            }
        }
        Date monthDayStart = DateUtil.getSomeMonthStart(new Date(), -1);
        String monthStartDate = DateUtil.format(monthDayStart, DATE_PATTERN.yyyyMMdd);
        LocalDate monthStart = LocalDate.parse(monthStartDate, dateTimeFormatter);
        long daysMonthDiff = ChronoUnit.DAYS.between(monthStart, end);
        List<String> months = monthRecord.stream().map(e -> e.getTime()).collect(Collectors.toList());
        for (int i = 0; i <= daysMonthDiff; i++) {
            String d = monthStart.plusDays(i).format(dateTimeFormatter);
            if (!months.contains(d)) {
                monthRecord.add(MonitorRecord.builder().time(d).count(0L).build());
            }
        }
        List<MonitorRecord> month =
            monthRecord.stream()
                .sorted(Comparator.comparing(MonitorRecord::getTime))
                .collect(Collectors.toList());
        List<MonitorRecord> week =
            weekRecord.stream()
                .sorted(Comparator.comparing(MonitorRecord::getTime))
                .collect(Collectors.toList());
        OfflineDevOperationTaskMonitorDomain offlineDevOperationTaskMonitorDomain =
            new OfflineDevOperationTaskMonitorDomain();
        offlineDevOperationTaskMonitorDomain.setMonthRecord(month);
        offlineDevOperationTaskMonitorDomain.setWeekRecord(week);
        return offlineDevOperationTaskMonitorDomain;
    }

    @Override
    public PageResult<OfflineDevOperationTaskMonitorErrorTaskDomain> getTaskMonitorErrorTask(
        Integer timeCase, Long projectId, Integer pageNum, Integer pageSize) {

        LambdaQueryWrapper<DatadevJobEntity> lambdaQueryWrapper =
            new LambdaQueryWrapper<DatadevJobEntity>();
        lambdaQueryWrapper
            .eq(DatadevJobEntity::getIsSubTask, 0)
            .eq(DatadevJobEntity::getProjectId, projectId)
            .eq(DatadevJobEntity::getExecType, 1)
            .in(
                DatadevJobEntity::getStatus,
                Lists.newArrayList(
                    OfflineDevJobStatusEnum.FAILURE.getCode(),
                    OfflineDevJobStatusEnum.PART_FAILED.getCode()));
        Calendar instance = Calendar.getInstance();
        instance.setTime(new Date());
        if (timeCase == 1) {
            instance.add(Calendar.DAY_OF_MONTH, -3);
        } else if (timeCase == 2) {
            instance.add(Calendar.DAY_OF_MONTH, -7);
        } else if (timeCase == 3) {
            instance.add(Calendar.MONTH, -1);
        } else if (timeCase == 4) {
            instance.add(Calendar.DAY_OF_MONTH, -1);
        }
        Date time = instance.getTime();
        lambdaQueryWrapper.ge(DatadevJobEntity::getEndTime, time);
        lambdaQueryWrapper.orderByDesc(DatadevJobEntity::getStartTime);
        IPage<DatadevJobEntity> page = this.page(new Page<>(pageNum, pageSize), lambdaQueryWrapper);
        List<DatadevJobEntity> list = page.getRecords();
        List<OfflineDevOperationTaskMonitorErrorTaskDomain> res =
            list.stream()
                .map(
                    e -> {
                        OfflineDevOperationTaskMonitorErrorTaskDomain convert =
                            Convert.convert(OfflineDevOperationTaskMonitorErrorTaskDomain.class, e);
                        convert.setTaskName(e.getNodeName());
                        convert.setTaskStatus(e.getStatus());
                        return convert;
                    })
                .collect(Collectors.toList());
        IPage<OfflineDevOperationTaskMonitorErrorTaskDomain> convert =
            page.convert(e -> Convert.convert(OfflineDevOperationTaskMonitorErrorTaskDomain.class, e));
        convert.setRecords(res);
        return PageResult.convert(convert);
    }

    @Override
    public Boolean saveParams(
        Long nodeId, List<OfflineDevTaskScriptParam> params, Long projectId, String userId) {

        List<DatadevSubNodeEntity> datadevSubNodeList =
            datadevSubNodeService.getDatadevSubNodeList(nodeId);

        // 工作流子节点Id
        List<Long> nodeIdList =
            datadevSubNodeList.stream()
                .map(DatadevSubNodeEntity::getNodeId)
                .collect(Collectors.toList());

        List<DatadevMenuNodeEntity> datadevMenuNodeEntities =
            datadevMenuNodeService.listByIds(nodeIdList);

        // 过滤出sql类型节点Id
        List<Long> sqlNodeIds =
            datadevMenuNodeEntities.stream()
                .filter(
                    datadevMenuNodeEntity ->
                        datadevMenuNodeEntity
                            .getNodeType()
                            .equals(OfflineDevNodeTypeEnum.SQL.getNodeType()))
                .map(EngineBaseModel::getId)
                .collect(Collectors.toList());

        // 遍历保存参数
        if (CollectionUtils.isNotEmpty(sqlNodeIds)) {
            for (Long sqlNodeId : sqlNodeIds) {
                offlineDevMenuNodeService.saveTaskScript(sqlNodeId, projectId, userId, null, params);
            }
        }

        return true;
    }

    @Override
    public Boolean ignoreDataDevJob(String ids) {
        if (StringUtils.isNotBlank(ids)) {
            List<Long> idList = Arrays.stream(ids.split(",")).map(e -> Long.valueOf(e))
                .collect(Collectors.toList());
            List<DatadevJobEntity> listByIds = this.listByIds(idList);
            for (DatadevJobEntity datadevJobEntity : listByIds) {
                if (datadevJobEntity.getNodeType()
                    .equals(OfflineDevNodeTypeEnum.DIR_TASK_FLOW.getNodeType())) {
                    List<DatadevJobEntity> list = this.list(
                        new LambdaQueryWrapper<>(DatadevJobEntity.class).eq(
                            DatadevJobEntity::getTaskFlowJobId, datadevJobEntity.getId()));
                    list.forEach(e -> {
                        e.setIgnored(1);
                        e.setStatus(OfflineDevJobStatusEnum.IGNORED.getCode());
                    });
                    this.updateBatchById(list);
                }
                datadevJobEntity.setStatus(OfflineDevJobStatusEnum.IGNORED.getCode());
                datadevJobEntity.setIgnored(1);
            }

            this.updateBatchById(listByIds);
            return true;
        }
        return true;
    }

    private Long getTaskFlowVnodeByTaskFlowId(Long taskFlowId) {
        List<DatadevSubNodeEntity> subTaskList =
            datadevSubNodeService.getDatadevSubNodeList(taskFlowId);

        for (DatadevSubNodeEntity subTask : subTaskList) {
            Long nodeId = subTask.getNodeId();
            DatadevMenuNodeEntity menuNode = datadevMenuNodeService.getById(nodeId);
            if (menuNode.getNodeType().equals(OfflineDevNodeTypeEnum.VNODE_START.getNodeType())) {
                return menuNode.getId();
            }
        }
        return null;
    }

    private Map<Long, OfflineDevBaseJobRunner> constructJobRunner(
        List<DatadevJobEntity> subJobList,
        List<DatadevMenuNodeEntity> menuNodeEntityList,
        List<DatadevSubNodeDependencyEntity> subTaskDependencyList) {
        Map<Long, OfflineDevBaseJobRunner> map = new HashMap<>();

        final Map<Long, DatadevMenuNodeEntity> nodeMap =
            menuNodeEntityList.stream().collect(Collectors.toMap(x -> x.getId(), x -> x));
        //
        //    final Map<Long, OfflineDevTaskConfigEntity> taskConfigEntityMap =
        //        taskConfigEntityList.stream().collect(Collectors.toMap(x -> x.getNodeId(), x -> x));

        final Map<Long, List<DatadevSubNodeDependencyEntity>> downUpsMap =
            subTaskDependencyList.stream()
                .collect(Collectors.groupingBy(DatadevSubNodeDependencyEntity::getDownNodeId));

        final Map<Long, List<DatadevSubNodeDependencyEntity>> upDownsMap =
            subTaskDependencyList.stream()
                .collect(Collectors.groupingBy(DatadevSubNodeDependencyEntity::getUpNodeId));

        for (DatadevJobEntity jobEntity : subJobList) {
            Long nodeId = jobEntity.getNodeId();

            OfflineDevBaseJobRunner jobRunner = getJobRunner(jobEntity, nodeMap.get(nodeId));

            if (null == jobRunner) {
                continue;
            }

            final List<DatadevSubNodeDependencyEntity> ups = downUpsMap.get(nodeId);

            if (CollectionUtils.isNotEmpty(ups)) {
                final Map<Long, Integer> parentNodeIdAndLineStatusMap =
                    ups.stream().collect(Collectors.toMap(x -> x.getUpNodeId(), x -> x.getStatus()));
                jobRunner.setParentNodeIdAndLineStatusMap(parentNodeIdAndLineStatusMap);
            }

            final List<DatadevSubNodeDependencyEntity> downs = upDownsMap.get(nodeId);

            if (CollectionUtils.isNotEmpty(downs)) {
                final List<Long> childNodeIdList =
                    downs.stream().map(d -> d.getDownNodeId()).collect(Collectors.toList());
                jobRunner.setChildNodeIdList(childNodeIdList);
            }

            map.put(nodeId, jobRunner);
        }

        return map;
    }

    private OfflineDevBaseJobRunner getJobRunner(
        DatadevJobEntity jobEntity, DatadevMenuNodeEntity nodeEntity) {

        final OfflineDevNodeTypeEnum nodeTypeEnum = OfflineDevNodeTypeEnum.of(nodeEntity.getNodeType());

        Map<Long, Integer> parentNodeIdAndLineStatusMap = new HashMap<>();
        List<Long> childNodeIdList = new ArrayList<>();
        Map<Long, OfflineDevBaseJobRunner> allJobNodeMap = new HashMap<>();

        switch (nodeTypeEnum) {
            case VNODE_START:
                return new OfflineDevVnodeJobRunner(
                    jobEntity,
                    nodeEntity,
                    parentNodeIdAndLineStatusMap,
                    childNodeIdList,
                    allJobNodeMap,
                    executorService);
            case SQL:
                return new OfflineDevSqlJobRunner(
                    jobEntity,
                    nodeEntity,
                    parentNodeIdAndLineStatusMap,
                    childNodeIdList,
                    allJobNodeMap,
                    executorService);
            case PYSPARK:
            case PYTHON2:
            case PYTHON3:
            case SHELL:
                return new OfflineDevShellJobRunner(
                    jobEntity,
                    nodeEntity,
                    parentNodeIdAndLineStatusMap,
                    childNodeIdList,
                    allJobNodeMap,
                    executorService);
            case OFFLINE:
                return new OfflineDevSyncTaskJobRunner(
                    jobEntity,
                    nodeEntity,
                    parentNodeIdAndLineStatusMap,
                    childNodeIdList,
                    allJobNodeMap,
                    executorService);
            case API:
                return new OfflineDevApiJobRunner(
                    jobEntity,
                    nodeEntity,
                    parentNodeIdAndLineStatusMap,
                    childNodeIdList,
                    allJobNodeMap,
                    executorService);
            case DATAQUALITY:
                return new OfflineDevDataQualityJobRunner(
                    jobEntity,
                    nodeEntity,
                    parentNodeIdAndLineStatusMap,
                    childNodeIdList,
                    allJobNodeMap,
                    executorService);
        }
        return null;
    }

    private OfflineDevBaseJobRunner getJobRunner(
        DatadevJobEntity jobEntity,
        DatadevMenuNodeEntity nodeEntity,
        DatadevScriptEntity datadevScriptEntity) {

        final OfflineDevNodeTypeEnum nodeTypeEnum = OfflineDevNodeTypeEnum.of(nodeEntity.getNodeType());

        Map<Long, Integer> parentNodeIdAndLineStatusMap = new HashMap<>();
        List<Long> childNodeIdList = new ArrayList<>();
        Map<Long, OfflineDevBaseJobRunner> allJobNodeMap = new HashMap<>();

        switch (nodeTypeEnum) {
            case VNODE_START:
                return new OfflineDevVnodeJobRunner(
                    jobEntity,
                    nodeEntity,
                    parentNodeIdAndLineStatusMap,
                    childNodeIdList,
                    allJobNodeMap,
                    executorService);
            case SQL:
                return new OfflineDevSqlJobRunner(
                    jobEntity,
                    nodeEntity,
                    parentNodeIdAndLineStatusMap,
                    childNodeIdList,
                    allJobNodeMap,
                    executorService,
                    datadevScriptEntity);
            case PYSPARK:
            case PYTHON2:
            case PYTHON3:
            case SHELL:
                String os = System.getProperty("os.name").toLowerCase();
                if (os.contains("win")) {
                    return new OfflineDevShellWinJobRunner(
                        jobEntity,
                        nodeEntity,
                        parentNodeIdAndLineStatusMap,
                        childNodeIdList,
                        allJobNodeMap,
                        executorService,
                        datadevScriptEntity);
                } else {
                    return new OfflineDevShellJobRunner(
                        jobEntity,
                        nodeEntity,
                        parentNodeIdAndLineStatusMap,
                        childNodeIdList,
                        allJobNodeMap,
                        executorService,
                        datadevScriptEntity);
                }
            case OFFLINE:
                return new OfflineDevSyncTaskJobRunner(
                    jobEntity,
                    nodeEntity,
                    parentNodeIdAndLineStatusMap,
                    childNodeIdList,
                    allJobNodeMap,
                    executorService);
            case API:
                return new OfflineDevApiJobRunner(
                    jobEntity,
                    nodeEntity,
                    parentNodeIdAndLineStatusMap,
                    childNodeIdList,
                    allJobNodeMap,
                    executorService);
            case DATAQUALITY:
                return new OfflineDevDataQualityJobRunner(
                    jobEntity,
                    nodeEntity,
                    parentNodeIdAndLineStatusMap,
                    childNodeIdList,
                    allJobNodeMap,
                    executorService);
        }
        return null;
    }

    @PreDestroy
    public void preDestroySetStatus() {
        String localHost = null;
        try {
            InetAddress inetAddress = InetAddress.getLocalHost();
            localHost = inetAddress.getHostAddress();
        } catch (UnknownHostException e) {
            log.info("获取本地ip失败{}", e.getMessage());
        }
        LambdaUpdateWrapper<DatadevJobEntity> set =
            new LambdaUpdateWrapper<DatadevJobEntity>()
                .eq(DatadevJobEntity::getStatus, OfflineDevJobStatusEnum.RUNNING.getCode())
                .set(DatadevJobEntity::getStatus, OfflineDevJobStatusEnum.CANCEL.getCode());
        if (StringUtils.isNotBlank(localHost)) {
            set.eq(DatadevJobEntity::getIpAddress, localHost);
        }
        this.update(set);
    }

    public static void main(String[] args) {
        System.out.println(new Date().toString());
    }
}
